/*
 * Copyright (c) 2018-2024 北京华恒盛世科技有限公司
 * QUICKPOOL is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *     http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
*/

#include "xdrbase.h"
#include "channel.h"
#include "limit.h"
#include "api.h"

#define ESUBNAME "usub"

static char *sstrUserMap;
static struct lenData sdataUsub = {0, NULL};

static JOBID_T send_batch(struct request *, struct lenData *,
                          struct QuickPoolSubmitReply *, struct poolMaster *);
static JOBID_T send_restart(struct request *, struct QuickPoolSubmitReply *, struct poolMaster *);
static int child_dothings(struct request *);

static void set_string_variable(FILE *, const char *, char *, unsigned int, unsigned int);
static void set_number_variable(FILE *, const char *, char *, int);
static void set_option_variable(FILE *, const char *, unsigned int, unsigned int, unsigned int);

static void set_restart_number_request(struct request *, unsigned int, char *);
static void set_restart_string_request(struct request *, unsigned int, char *);

static int handle_other_files(char *, char *, struct request *, unsigned int *);
static char *unwrap_command(char *);
static void handle_command_argument(char *, struct request *, enum submitType, const char *);
static void handle_string_argument(char *, struct request *, unsigned int, unsigned int,
                                   unsigned int *, const char *);
static void handle_number_argument(char *, struct request *, unsigned int, unsigned int,
                                   unsigned int *, const char *);
static void handle_option_argument(char *, unsigned int, unsigned int *,
                                   unsigned int *, const char *);
static int is_stringtoken(char *, const char *);

/**
 * @brief       API: submit a job to the cluster.
 *
 * @param[in]   req          #1: job submit request key values
 * @param[out]  submitreply  #2: submit job reply from JS
 * @param[in]   master       #3: master(port and name) to be called
 *
 * @retval      > 0          #1: succeed, got job ID from JS
 * @retval      -1           #2: failed
 *
 * @note        caller should calloc memory for QuickPoolSubmitReply if need.
 *              For JS replied error, caller should handle it.
 */
JOBID_T
client_submit_job(struct request *req, struct QuickPoolSubmitReply *submitrep,
                  struct poolMaster *master)
{
    unsigned int options=0, options2=0, optIndex=0, opt2Index=0;
    char *queue=NULL, *cwd=NULL, *chkpntDir=NULL, *chkpntMethod=NULL, *project=NULL, *command=NULL;
    unsigned int i;
    time_t beginTime=0, termTime=0;
    enum submitType submit=CMD_SUBMIT_JOB;
    int numProcessors=0, maxProcessors=0, signalValue=0;
    int submitTimeFlag = FALSE;
    int cwdIndex=-1;
    for (i=0; i<req->number; i++) {
        switch (req->keyvalues[i].key) {
        case REQUEST_KEY_SUBMIT_RESREQ:
        case REQUEST_KEY_SUBMIT_PRECMD:
        case REQUEST_KEY_SUBMIT_POSTCMD:    
        case REQUEST_KEY_SUBMIT_DEPENDENCY:
        case REQUEST_KEY_SUBMIT_JOBNAME:
        case REQUEST_KEY_SUBMIT_ERRFILE:
        case REQUEST_KEY_SUBMIT_INFILE:
        case REQUEST_KEY_SUBMIT_OUTFILE:
        case REQUEST_KEY_SUBMIT_XFILE:
        case REQUEST_KEY_SUBMIT_NODE:
            strip_newline(req->keyvalues[i].value);
            break;
        case REQUEST_KEY_SUBMIT_TYPE:
            submit = (enum submitType)atoi(req->keyvalues[i].value);
            break;
        case REQUEST_KEY_SUBMIT_QUEUE:
            queue = req->keyvalues[i].value;        
            break;
        case REQUEST_KEY_SUBMIT_PROJECT:
            project = req->keyvalues[i].value;
            break;
        case REQUEST_KEY_SUBMIT_COMMAND:
            command = req->keyvalues[i].value;
            break;
        case REQUEST_KEY_SUBMIT_BEGINTIME:
            beginTime = atoi(req->keyvalues[i].value);
            break;
        case REQUEST_KEY_SUBMIT_TERMTIME:
            termTime = atoi(req->keyvalues[i].value);
            break;
        case REQUEST_KEY_SUBMIT_NUM_PROCESSOR:
            numProcessors = atoi(req->keyvalues[i].value);
            break;
        case REQUEST_KEY_SUBMIT_MAX_PROCESSOR:
            maxProcessors = atoi(req->keyvalues[i].value);
            break;
        case REQUEST_KEY_SUBMIT_SIGNAL:
            signalValue = atoi(req->keyvalues[i].value);
            break;
        case REQUEST_KEY_SUBMIT_CWD:
            cwd = req->keyvalues[i].value; 
            cwdIndex = i;
            break;
        case REQUEST_KEY_SUBMIT_OPTIONS:
            options = atoi(req->keyvalues[i].value);
            optIndex = i;
            break;
        case REQUEST_KEY_SUBMIT_OPTIONS2:
            options2 = atoi(req->keyvalues[i].value);
            opt2Index = i;
            break;
        case REQUEST_KEY_CHKPNT_DIR:
            strip_newline(req->keyvalues[i].value);
            chkpntDir = req->keyvalues[i].value;
            break;
        case REQUEST_KEY_CHKPNT_METHOD:
            strip_newline(req->keyvalues[i].value);
            chkpntMethod = req->keyvalues[i].value;
            break;
        case REQUEST_KEY_SUBMIT_TIME:
            submitTimeFlag = TRUE;
            break;
       }
    }

    unsigned int n = req->number;
    if (NULL == queue) { /* -q is not specified, use QP_DEFAULTQUEUE variable */

        queue = getenv("QP_DEFAULTQUEUE");
        if (NULL != queue && '\0' != queue[0]) {

            req->keyvalues[n].key = REQUEST_KEY_SUBMIT_QUEUE;
            req->keyvalues[n].value = copy_string(queue);
            n++;
        }
    }

    struct group *grpEntry;
    grpEntry = getgrgid(getgid());
    if (NULL != grpEntry) {
        gf_setenv("QP_UNIXGROUP", grpEntry->gr_name);
    }

    unlink_delta_files();

    if (NULL == cwd) {
        char mycwd[PATH_MAX];

        if (NULL == gf_getwd(mycwd)) {
            return -1;
        }

        if (0 <= cwdIndex) {
            req->keyvalues[cwdIndex].value = copy_string(mycwd);
        } else {           
            req->keyvalues[n].key = REQUEST_KEY_SUBMIT_CWD;
            req->keyvalues[n].value = copy_string(mycwd);
            n++;            
        }
    }

    req->number = n;
    if (!(options & SUBOPT_NOFORK)) {
        if (child_dothings(req) < 0) { // usub to modify user submit request.
            return -1;
        }
    }

    change_submit_request(req, submit); // change request if esub modify some.

    options |= SUBOPT_UNIT_KB;

    if (numProcessors < 0) {
        log_message(ERR, NOCLASS, "%s: num processor should be a positive integer");
        return -1;
    }
    if (maxProcessors < numProcessors) {
        log_message(ERR, NOCLASS, "%s: max processor should be bigger than num processor.",
                    __func__);
        return -1;
    }

    if (numProcessors == 0 && maxProcessors == 0) {

        options2 |= SUBOPT2_DEF_PROCLIMIT;
    }

    if (beginTime > 0 && termTime > 0 && beginTime > termTime) {
        log_message(ERR, NOCLASS, "%s: job terminate time should be later than job begin time.",
                    __func__);
        return -1;
    }

    if (signalValue > _NSIG || signalValue < 0) {
        log_message(ERR, NOCLASS, "%s: unknown signal value[%d].", __func__, signalValue);
        return -1;
    }

    if (NULL != chkpntDir) { // chkpnt job

        if (NULL == chkpntMethod) { // not specify chkpnt method, use parameter or default.
            if (baseParams[QP_CHKPNT_METHOD].string_value != NULL) {
                chkpntMethod = baseParams[QP_CHKPNT_METHOD].string_value; 
            } else {
                chkpntMethod = DEFAULT_STRING;
            }

            req->keyvalues[n].key = REQUEST_KEY_CHKPNT_METHOD;
            req->keyvalues[n].value = copy_string(chkpntMethod);
            n++;
        }
    }

    if (!submitTimeFlag) {
        req->keyvalues[n].key = REQUEST_KEY_SUBMIT_TIME;
        req->keyvalues[n].value = (char *)calloc(12, sizeof(char));
        if (NULL == req->keyvalues[n].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            return -1;
        }
        snprintf(req->keyvalues[n].value, 12, "%ld", time(NULL));
        n++;
    }
    
    if (0 < optIndex) {
        snprintf(req->keyvalues[optIndex].value, 12, "%d", options);
    } else {
        req->keyvalues[n].key = REQUEST_KEY_SUBMIT_OPTIONS;
        req->keyvalues[n].value = (char *)calloc(12, sizeof(char));
        if (NULL == req->keyvalues[n].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            return -1;
        }
        snprintf(req->keyvalues[n].value, 12, "%d", options);
        n++;
    }

    if (0 < opt2Index) {
        snprintf(req->keyvalues[opt2Index].value, 12, "%d", options2);
    } else {
        req->keyvalues[n].key = REQUEST_KEY_SUBMIT_OPTIONS2;
        req->keyvalues[n].value = (char *)calloc(12, sizeof(char));
        if (NULL == req->keyvalues[n].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            return -1;
        }
        snprintf(req->keyvalues[n].value, 12, "%d", options2);
        n++;
    }

    JOBID_T jobId = -1;
    req->number = n;
    if (0 > set_more_request(req, submit)) {
        return jobId;
    }
    n = req->number;
    
    struct lenData jf;
    if (0 > set_jobfile_data(project, command, &jf)) {
        return jobId;
    }

    req->number = n;
    if (submit == CMD_RESTART_JOB) { // restart job
        jobId = send_restart(req, submitrep, master);
    } else { // submit job
        jobId = send_batch(req, &jf, submitrep, master);
    }
    free(jf.data);
    
    return jobId;
} // end function client_submit_job

#define MSG_BAD_ENVAR2s ("%s: Bad parameter variable name(%s) " \
                              "read from $QP_SUBMIT_MODIFY_FILE, the " \
                              "setting will be ignored.")

/**
 * @brief       read the delta file and change job submit request accordingly.
 *
 * @param[in|out]   req      #1: job submit request key values
 * @param[in]       submit   #2: submit, modify or restart a job
 *
 * @note        the delta file is created by usub script.
 *              do not update submit request if the file is not exist.
 */
void
change_submit_request(struct request *req, enum submitType submit)
{
    static struct {
        const char *name;
        unsigned int option;
        int key;
    } reqParams[]= {
        {"QP_SUBMIT_JOB_NAME", SUBOPT_JOB_NAME, REQUEST_KEY_SUBMIT_JOBNAME},
        {"QP_SUBMIT_QUEUE", SUBOPT_QUEUE, REQUEST_KEY_SUBMIT_QUEUE},
        {"QP_SUBMIT_IN_FILE", SUBOPT_IN_FILE, REQUEST_KEY_SUBMIT_INFILE},
        {"QP_SUBMIT_OUT_FILE", SUBOPT_OUT_FILE, REQUEST_KEY_SUBMIT_OUTFILE},
        {"QP_SUBMIT_ERR_FILE", SUBOPT_OUT_FILE, REQUEST_KEY_SUBMIT_ERRFILE},
        {"QP_SUBMIT_EXCLUSIVE", SUBOPT_EXCLUSIVE, -1},
        {"QP_SUBMIT_NOTIFY_END", SUBOPT_NOTIFY_END, -1},
        {"QP_SUBMIT_OUT_NOTIFY_BEGIN", SUBOPT_NOTIFY_BEGIN, -1},
        {"QP_SUBMIT_RERUNNABLE", SUBOPT_RERUNNABLE, -1},
        {"QP_SUBMIT_WINDOW_SIG", SUBOPT_WINDOW_SIG, REQUEST_KEY_SUBMIT_SIGNAL},
        {"QP_SUBMIT_DEPEND_COND", SUBOPT_DEPEND_COND, REQUEST_KEY_SUBMIT_DEPENDENCY},
        {"QP_SUBMIT_RES_REQ", SUBOPT_RES_REQ, REQUEST_KEY_SUBMIT_RESREQ},
        {"QP_SUBMIT_PRE_EXEC", SUBOPT_PRE_EXEC, REQUEST_KEY_SUBMIT_PRECMD},
        {"QP_SUBMIT_POST_EXEC", SUBOPT_POST_EXEC, REQUEST_KEY_SUBMIT_POSTCMD},
        {"QP_SUBMIT_LOGIN_SHELL", SUBOPT_LOGIN_SHELL, REQUEST_KEY_SUBMIT_LOGINSHELL},
        {"QP_SUBMIT_MAIL_USER", SUBOPT_MAIL_USER, REQUEST_KEY_SUBMIT_MAIL},
        {"QP_SUBMIT_MODIFY", SUBOPT_MODIFY, REQUEST_KEY_SUBMIT_TYPE},
        {"QP_SUBMIT_PROJECT_NAME", SUBOPT_PROJECT_NAME, REQUEST_KEY_SUBMIT_PROJECT},
        {"QP_SUBMIT_HOSTS", SUBOPT_HOST, REQUEST_KEY_SUBMIT_NODE},
        {"QP_SUBMIT_HOLD", SUBOPT2_HOLD, -2},
        {"QP_SUBMIT_JOB_PRIORITY", SUBOPT_JOB_PRIORITY, REQUEST_KEY_SUBMIT_JOBPRIORITY},
        {"QP_SUBMIT_RLIMIT_CPU", 0, REQUEST_KEY_SUBMIT_CPULIMIT},
        {"QP_SUBMIT_RLIMIT_FSIZE", 0, REQUEST_KEY_SUBMIT_FILELIMIT},
        {"QP_SUBMIT_RLIMIT_DATA", 0, REQUEST_KEY_SUBMIT_DATALIMIT},
        {"QP_SUBMIT_RLIMIT_STACK", 0, REQUEST_KEY_SUBMIT_STACKLIMIT},
        {"QP_SUBMIT_RLIMIT_CORE", 0, REQUEST_KEY_SUBMIT_CORELIMIT},
        {"QP_SUBMIT_RLIMIT_RSS", 0, REQUEST_KEY_SUBMIT_RSSLIMIT},
        {"QP_SUBMIT_RLIMIT_SWAP", 0, REQUEST_KEY_SUBMIT_SWAPLIMIT},
        {"QP_SUBMIT_RLIMIT_RUN", 0, REQUEST_KEY_SUBMIT_RUNLIMIT},
        {"QP_SUBMIT_RLIMIT_PROCESS", 0, REQUEST_KEY_SUBMIT_PROCESSLIMIT},
        {"QP_SUBMIT_RLIMIT_NOFILE", 0, REQUEST_KEY_SUBMIT_NOFILELIMIT},
        {"QP_SUBMIT_RLIMIT_OPEN_MAX", 0, REQUEST_KEY_SUBMIT_OPENMAXLIMIT},
        {"QP_SUBMIT_NUM_PROCESSORS", 0, REQUEST_KEY_SUBMIT_NUM_PROCESSOR},
        {"QP_SUBMIT_MAX_NUM_PROCESSORS", 0, REQUEST_KEY_SUBMIT_MAX_PROCESSOR},
        {"QP_SUBMIT_BEGIN_TIME", 0, REQUEST_KEY_SUBMIT_BEGINTIME},
        {"QP_SUBMIT_TERM_TIME", 0, REQUEST_KEY_SUBMIT_TERMTIME},
        {NULL, 0},
    };

    char strParamDeltaFile[PATH_MAX+1];
    snprintf(strParamDeltaFile, PATH_MAX, "%s/.submit_delta_parameter.%d.%d",
             gstrTmpDir, getpid(), getuid());

    unsigned int lineNum;

    FILE *fileParam;
    fileParam = fopen(strParamDeltaFile,"r");
    if (NULL != fileParam) {
        unsigned int i;
        unsigned int options=0, options2=0, optIndex=0, opt2Index=0;
        unsigned int deleteOptions=0, deleteOptions2=0, delOptIndex=0, delOpt2Index=0;
        for (i=0; i<req->number; i++) {
            switch (req->keyvalues[i].key) {
            case REQUEST_KEY_SUBMIT_OPTIONS:
                options = atoi(req->keyvalues[i].value);
                optIndex = i;
                break;
            case REQUEST_KEY_SUBMIT_OPTIONS2:
                options2 = atoi(req->keyvalues[i].value);
                opt2Index = i;
                break;
            case REQUEST_KEY_MODIFY_DELETE_OPTIONS:
                deleteOptions = atoi(req->keyvalues[i].value);
                delOptIndex = i;
                break;
            case REQUEST_KEY_MODIFY_DELETE_OPTIONS2:
                deleteOptions2 = atoi(req->keyvalues[i].value);
                delOpt2Index = i;
                break;
            }
        }

        lineNum=0;

        char buf[BUFSIZ];
        char *line;
        while (NULL != (line=gf_get_line(fileParam, &lineNum, buf))) {
            int i;
            unsigned int len = strlen(line);

            char *key, *token;
            key = (char *)calloc(len+1, sizeof(char));
            if (NULL == key) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", len+1);
                return;
            }

            token = get_string_token(&line, " \t=!@#$%^&*()", key, len+1);
            if (NULL == token) {
                FREEUP(key);
                continue;
            }

            while(*line!='=') line++;
            line++;
            line = strip_spaces(line, FALSE);

            len = strlen("QP_SUBMIT_OTHER_FILES");
            if (0 == strncmp(token, "QP_SUBMIT_OTHER_FILES", len)) {
                handle_other_files(token, line, req, &deleteOptions);
                FREEUP(key);
                continue;
            }

            if (0 == strcmp(token, "QP_SUBMIT_COMMAND_LINE") && CMD_RESTART_JOB == submit) {
                FREEUP(key);
                continue;
            } else if (0 == strcmp(token, "QP_SUBMIT_COMMAND_LINE")) {
                handle_command_argument(line, req, submit, token);
                continue;
            }

            bool_t foundKey = FALSE;
            for(i=0; reqParams[i].name; i++) {

                if (0 != strcmp(token, reqParams[i].name)) {
                    continue;
                }

                foundKey = TRUE;

                switch (reqParams[i].key) {
                case REQUEST_KEY_SUBMIT_JOBNAME:
                case REQUEST_KEY_SUBMIT_QUEUE:
                case REQUEST_KEY_SUBMIT_INFILE:
                case REQUEST_KEY_SUBMIT_OUTFILE:
                case REQUEST_KEY_SUBMIT_ERRFILE:
                case REQUEST_KEY_SUBMIT_SIGNAL:
                case REQUEST_KEY_SUBMIT_DEPENDENCY:
                case REQUEST_KEY_SUBMIT_RESREQ:
                case REQUEST_KEY_SUBMIT_PRECMD:
                case REQUEST_KEY_SUBMIT_POSTCMD:    
                case REQUEST_KEY_SUBMIT_LOGINSHELL:
                case REQUEST_KEY_SUBMIT_MAIL:
                case REQUEST_KEY_SUBMIT_PROJECT:
                case REQUEST_KEY_SUBMIT_NODE:

                    handle_string_argument(line, req, reqParams[i].key, reqParams[i].option,
                                           &deleteOptions, token);
                    break;
                case REQUEST_KEY_SUBMIT_CPULIMIT:
                case REQUEST_KEY_SUBMIT_RUNLIMIT:
                case REQUEST_KEY_SUBMIT_FILELIMIT:
                case REQUEST_KEY_SUBMIT_DATALIMIT:
                case REQUEST_KEY_SUBMIT_STACKLIMIT:
                case REQUEST_KEY_SUBMIT_CORELIMIT:
                case REQUEST_KEY_SUBMIT_RSSLIMIT:
                case REQUEST_KEY_SUBMIT_SWAPLIMIT:
                case REQUEST_KEY_SUBMIT_PROCESSLIMIT:
                case REQUEST_KEY_SUBMIT_NOFILELIMIT:
                case REQUEST_KEY_SUBMIT_OPENMAXLIMIT:
                case REQUEST_KEY_SUBMIT_BEGINTIME:
                case REQUEST_KEY_SUBMIT_TERMTIME:
                case REQUEST_KEY_SUBMIT_NUM_PROCESSOR:
                case REQUEST_KEY_SUBMIT_MAX_PROCESSOR:
                case REQUEST_KEY_SUBMIT_JOBPRIORITY:

                    handle_number_argument(line, req, reqParams[i].key, reqParams[i].option,
                                           &deleteOptions2, token);
                    break;
                case -1:

                    handle_option_argument(line, reqParams[i].option, &options,
                                           &deleteOptions, token);
                    break;
                case -2:

                    handle_option_argument(line, reqParams[i].option, &options2,
                                           &deleteOptions2, token);
                    break;
                default:
                    log_message(WARNING, NOCLASS, MSG_BAD_ENVAR2s, __func__, token);
                    break;
                }
                break;
            }

            if (!foundKey) {
                log_message(WARNING, NOCLASS, MSG_BAD_ENVAR2s, __func__, token);
            }

            FREEUP(key);
        }

        if (0 <= optIndex) {
            snprintf(req->keyvalues[optIndex].value, 12, "%d", options);
        } else if (0 < options) {
            req->keyvalues[req->number].value = (char *)calloc(12, sizeof(char));
            if (NULL == req->keyvalues[req->number].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                return;
            }
            snprintf(req->keyvalues[req->number].value, 12, "%d", options);
            req->number ++;
        }

        if (0 <= opt2Index) {
            snprintf(req->keyvalues[opt2Index].value, 12, "%d", options2);
        } else if (0 < options2) {
            req->keyvalues[req->number].value = (char *)calloc(12, sizeof(char));
            if (NULL == req->keyvalues[req->number].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                return;
            }
            snprintf(req->keyvalues[req->number].value, 12, "%d", options2);
            req->number ++;
        }

        if (0 <= delOptIndex) {
            snprintf(req->keyvalues[delOptIndex].value, 12, "%d", deleteOptions);
        } else if (0 < deleteOptions) {
            req->keyvalues[req->number].value = (char *)calloc(12, sizeof(char));
            if (NULL == req->keyvalues[req->number].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                return;
            }
            snprintf(req->keyvalues[req->number].value, 12, "%d", deleteOptions);
            req->number ++;
        }

        if (0 <= delOpt2Index) {
            snprintf(req->keyvalues[delOptIndex].value, 12, "%d", deleteOptions2);
        } else if (0 < deleteOptions2) {
            req->keyvalues[req->number].value = (char *)calloc(12, sizeof(char));
            if (NULL == req->keyvalues[req->number].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                return;
            }
            snprintf(req->keyvalues[req->number].value, 12, "%d", deleteOptions2);
            req->number ++;
        }

        fclose(fileParam);
        unlink(strParamDeltaFile);
    }

    char strEnvDeltaFile[PATH_MAX+1];
    snprintf(strEnvDeltaFile, PATH_MAX, "%s/.submit_delta_env.%d.%d",
             gstrTmpDir, getpid(), getuid());
    if(F_OK == access(strEnvDeltaFile, R_OK)) {

        FILE *fileEnv;
        fileEnv = fopen(strEnvDeltaFile, "r");

        if (NULL != fileEnv) {
            char buf[BUFSIZ];
            char *line;

            lineNum=0;
            while (NULL != (line=gf_get_line(fileEnv, &lineNum, buf))) {

                unsigned int len = strlen(line);
                char *key, *token;
                key = (char *)calloc(len+1, sizeof(char));
                if (NULL == key) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", len+1);
                    return;
                }

                token = get_string_token(&line, " \t =!@#$%^&*()", key, len+1);
                if (NULL == token) {
                    FREEUP(key);
                    continue;
                }

                while(*line!='=') line++;

                line++;

                char *sp = get_string_inside(&line, '"', '"');
                gf_setenv(token, sp);

                FREEUP(sp);
                FREEUP(key);
            }

            fclose(fileEnv);
        }

        unlink(strEnvDeltaFile);
    }

    return;
} // end function change_submit_request

/**
 * @brief       put enveironment variables and job command into job file data.
 *
 * @param[in]   project      #1: submit project name
 * @param[in]   command      #2: job command
 * @param[out]  jf           #3: job file data
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: not enough memory
 *
 * @note        bypass some QP_xxx environment variables.
 */
int
set_jobfile_data(char *project, char *command, struct lenData *jf)
{
    int commandlen = 0;
    if (NULL != command) {
        commandlen = strlen(command);
    }

    if (NULL != project) {
        setenv("QP_PROJECT_NAME", project, 0);
    }

    int length;
    length = sizeof(CMDSTART) + sizeof(TRAPSIGCMD) + sizeof(WAITCLEANCMD);
    length += sizeof(EXITCMD) + commandlen + sizeof(NUMENV) + sizeof(ENVSSTART);
    length += sizeof(EDATASTART) + sizeof(SHELLLINE) + 1 + MAXNAMELEN*2 + sdataUsub.len;

    jf->len = 0;

    int size;
    size = MAX(length, MAXMESSAGELENGTH);

    jf->data = (char *)calloc(size, sizeof(char));
    if (NULL == jf->data) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", size);
        return -1;
    }
    jf->data[0] = '\0';
    strcat(jf->data, SHELLLINE);

    if (NULL != sstrUserMap) {
        strcat(jf->data, "QP_ACCT_MAP='");
        strcat(jf->data, sstrUserMap);
        strcat(jf->data, "'; export QP_ACCT_MAP\n");
        length += 14 + strlen(sstrUserMap) + 24;
        FREEUP(sstrUserMap);
    }

    char num[MAXNAMELEN+1];
    strcat(jf->data, "QUICKPOOL_VERSION='");
    snprintf(num, MAXNAMELEN, "%d", QUICKPOOL_VERSION);
    strcat(jf->data, num);
    strcat(jf->data, "'; export QUICKPOOL_VERSION\n");
    length += 13 + strlen(num) + 23;

    char **ep;
    for (ep = environ; *ep; ep++) {

        if (!strncmp(*ep, "QP_JOBID=", 9) || !strncmp(*ep, "QP_HOSTS=", 9)
            || !strncmp(*ep, "QP_QUEUE=", 9) || !strncmp(*ep, "QP_JOBNAME=", 11)
            || !strncmp(*ep, "QP_TRAPSIGS=", 12) || !strncmp(*ep, "QP_JOBFILENAME=", 15)
            || !strncmp(*ep, "QP_RESTART=", 11) || !strncmp(*ep, "QP_EXIT_PRE_ABORT=", 18)
            || !strncmp(*ep, "QP_EXIT_REQUEUE=", 16) || !strncmp(*ep, "QP_JOBPID=", 10)
            || !strncmp(*ep, "QP_SUBCWD=", 10) || !strncmp(*ep, "QP_ACCT_MAP=", 13)
            || !strncmp(*ep, "QP_JOB_STARTER=", 15) || !strncmp(*ep, "QUICKPOOL_VERSION=", 18)
            || !strncmp(*ep, "QP_SUBMIT_", 10)
            /* jm sets these variables */
            || !strncmp(*ep, "HOME=", 5) || !strncmp(*ep, "PWD=", 4) || !strncmp(*ep, "USER=", 5)) {
            continue;
        }

        if (!strncmp(*ep, "TERMCAP=", 8) || !strncmp(*ep, "TERM=", 5)) {
            continue;
        }

        char *strEnv, *keepEnv;
        strEnv = copy_string(*ep);
        keepEnv = strEnv;

        char *strEqual;
        for (strEqual=strEnv; *strEqual!='\0' && *strEqual!='='; strEqual++);

        bool_t noEqual=FALSE;
        if ('\0' == *strEqual) {

            noEqual = TRUE;
        } else {
            *strEqual = '\0';
        }

        int len, len1;
        if (noEqual) {
            len1 = 2;
        } else {
            len1 = strlen(strEqual+1) + 1;
        }

        len = strlen(strEnv) + len1 + sizeof(TAILCMD) + strlen(strEnv) + 1;
        if ((length += len) > size) {

            size += (len > MAXMESSAGELENGTH ? len : MAXMESSAGELENGTH);
            char *newp = (char *)realloc(jf->data, size);
            if (NULL == newp) {
                if (!noEqual) {
                    *strEqual = '=';
                }

                log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "realloc" );
                FREEUP(keepEnv);
                free(jf->data);
                return -1;
            }

            jf->data = newp;
        }
        strcat(jf->data, strEnv);
        strcat(jf->data, "='");

        if (noEqual) {
            strcat(jf->data, "\0");
        } else {
            strcat(jf->data, strEqual+1);
        }

        strcat(jf->data, TAILCMD);
        strcat(jf->data, strEnv);
        strcat(jf->data, "\n");
        if (!noEqual) {
            *strEqual = '=';
        }

        FREEUP(keepEnv);
    }

    int len;
    len = sizeof(TRAPSIGCMD) + sizeof(CMDSTART) + commandlen + sizeof(WAITCLEANCMD)
          + sizeof(EXITCMD) + sizeof(EDATASTART) + sdataUsub.len + MAXNAMELEN;
    if ((length += len) > size) {

        size += (len > MAXMESSAGELENGTH ? len : MAXMESSAGELENGTH);
        char *newp = (char *)realloc(jf->data, size);

        if (newp == NULL) {
            log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "realloc" );
            FREEUP(jf->data);
            return -1;
        }
        jf->data = newp;
    }

    log_message (DEBUG2, TRACER, "%s:length=%d, size=%d, jf->len=%d.",
                 __func__, length, size, strlen(jf->data));

    strcat(jf->data, TRAPSIGCMD);
    strcat(jf->data, CMDSTART);

    if (command) {
        strcat(jf->data, command);
    }

    strcat(jf->data, WAITCLEANCMD);
    strcat(jf->data, EXITCMD);

    strcat(jf->data, EDATASTART);
    snprintf(num, MAXNAMELEN, "%d\n", sdataUsub.len);
    strcat(jf->data, num);

    char *strExtra;
    strExtra = jf->data + strlen(jf->data) + 1;
    memcpy(strExtra, sdataUsub.data, sdataUsub.len);
    jf->len = strlen(jf->data) + 1 + sdataUsub.len;

    return 0;
} // end function set_jobfile_data

/**
 * @brief       after fill_submit_request, add some more info to submit request.
 *
 * @param[out]  req          #1: request to be set more request
 * @param[in]   submit       #2: submit, modify or restart a job
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed
 */
int
set_more_request(struct request *req, enum submitType submit)
{
    unsigned int i;
    char *command=NULL, *chkpntDir=NULL, *cwd=NULL;
    char *fromHost = NULL, *userNmae = NULL, *homeDir = NULL;
    int userIdFlag = FALSE, uMaskFlag = FALSE;
    for (i=0; i<req->number; i++) {
        switch (req->keyvalues[i].key) {
        case REQUEST_KEY_MODIFY_COMMAND:
            command = req->keyvalues[i].value;
            break;
        case REQUEST_KEY_SUBMIT_COMMAND:
            command = req->keyvalues[i].value;
            break;
        case REQUEST_KEY_RESTART_CHKPNTDIR:
            chkpntDir = req->keyvalues[i].value;
            break;
        case REQUEST_KEY_SUBMIT_CWD:
            cwd = req->keyvalues[i].value;
            break;
        case REQUEST_KEY_SUBMIT_FROMHOST:
            fromHost = req->keyvalues[i].value;
            break;
        case REQUEST_KEY_SUBMIT_UMASK:
            uMaskFlag = TRUE;
            break;
        case REQUEST_KEY_SUBMIT_HOMEDIR:
            homeDir = req->keyvalues[i].value;
            break;
        case REQUEST_KEY_SUBMIT_USERID:
            userIdFlag = TRUE;
            break;
        case REQUEST_KEY_SUBMIT_USERNAME:
            userNmae = req->keyvalues[i].value;
            break;           
        }
    }

    unsigned int n = req->number;

    if (CMD_SUBMIT_JOB == submit && NULL == command) {

        log_message(ERR, NOCLASS, "%s: job command is not specified.", __func__);
        return -1;
    } else if (CMD_RESTART_JOB == submit && NULL == chkpntDir) {

        log_message(ERR, NOCLASS, "%s: checkpoint directory is not specified.", __func__);
        return -1;
    }
    
    if (NULL != command) {
        char *lineStrBuf, *jobdespBuf, lastNonSpaceChar, *jobdesp, *sp;
        int lastNonSpaceIdx, reqCmd1Offset;
        lineStrBuf = copy_string(command);
        jobdesp = lineStrBuf;

        if ((sp = strstr(jobdesp, "SCRIPT_\n")) != NULL) {

            reqCmd1Offset = sp + strlen("SCRIPT_\n") - jobdesp;
            jobdesp += reqCmd1Offset;
            if ((sp = strstr(jobdesp, "SCRIPT_\n")) != NULL) {
                while (*sp != '\n') --sp;
                *sp = '\0';
            }
        } else {
            reqCmd1Offset = 0;
        }

        jobdespBuf = copy_string(jobdesp);
        jobdesp = jobdespBuf;
        FREEUP(lineStrBuf);

        for (lastNonSpaceIdx = 0, lastNonSpaceChar = ';', i = 0;
             jobdesp[i] != '\0'; i++) {

            if ('\n' == jobdesp[i]) {
                if (';' != lastNonSpaceChar && '&' != lastNonSpaceChar) {
                    jobdesp[i] = ';';
                } else {
                    jobdesp[i] = ' ';
                }
            }

            if (jobdesp[i] != ' ' && jobdesp[i] != '\t') {
                lastNonSpaceChar = jobdesp[i];
                lastNonSpaceIdx = i;
            }
        }

        if (';' == jobdesp[lastNonSpaceIdx]) {
            jobdesp[lastNonSpaceIdx] = '\0';
        }

        for (i = 0; command[i] != '\0'; i++) {

            if (';' == command[i] || ' ' == command[i] || '&' == command[i]
                || '>' == command[i] || '<' == command[i] || '|' == command[i]
                || '\t' == command[i] || '\n' == command[i]) {
                break;
            }
        }

        if (0 == i && ' ' != command[0]) {
            log_message(ERR, NOCLASS, "%s: job command <%s> is invalid.", __func__, command);
            return -1;
        }

        strcpy(command, jobdesp);
        FREEUP(jobdespBuf);
    }

    char *myHostName;
    myHostName = qp_getmyhostname();
    if (NULL == myHostName) {
        log_message(ERR, NOCLASS, STR_FUNC_FAIL, __func__, "qp_getmyhostname");
        return -1;
    }

    if (NULL == fromHost) {
        req->keyvalues[n].key = REQUEST_KEY_SUBMIT_FROMHOST;
        req->keyvalues[n].value = copy_string(myHostName);
        n++;
    }

    if (!uMaskFlag) {
        int usermask = umask(0077);
        req->keyvalues[n].key = REQUEST_KEY_SUBMIT_UMASK;
        req->keyvalues[n].value = (char *)calloc(12, sizeof(char));
        if (NULL == req->keyvalues[n].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            return -1;
        }
        snprintf(req->keyvalues[n].value, 12, "%d", usermask);
        n++;
    }
    
    char uName[MAXNAMELEN];
    struct passwd pw, *result;
    getpwuid_r(getuid(), &pw, uName, sizeof(uName), &result);
    if (NULL == result) {
        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "getpwuid_r" );
        return -1;
    }

    if (strlen(pw.pw_dir) >= PATH_MAX - 1) {
        log_message(ERR, NOCLASS, "%s: pw_dir is longer than %d.", __func__, PATH_MAX);
        errno = ENAMETOOLONG;
        return -1;
    }

    if (NULL != cwd) {
        for (i = 0; pw.pw_dir[i] != '\0' && cwd != NULL && cwd[i] == pw.pw_dir[i]; i++);

        if ((i != 0) && (pw.pw_dir[i] == '\0')) {
            if (cwd[i] == '\0') {
                cwd[0] = '\0';
            } else if (cwd[i] == '/') {
                strcpy(cwd, cwd+i+1);
            }
        }
    }

    if (NULL == homeDir) {
        req->keyvalues[n].key = REQUEST_KEY_SUBMIT_HOMEDIR;
        req->keyvalues[n].value = copy_string(pw.pw_dir);
        n++;
    }

    if (!userIdFlag) {
        req->keyvalues[n].key = REQUEST_KEY_SUBMIT_USERID;
        req->keyvalues[n].value = (char *)calloc(12, sizeof(char));
        if (NULL == req->keyvalues[n].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            return -1;
        }
        snprintf(req->keyvalues[n].value, 12, "%d", getuid());
        n++;
    }

    if (NULL == userNmae) {
        req->keyvalues[n].key = REQUEST_KEY_SUBMIT_USERNAME;
        req->keyvalues[n].value = copy_string(uName);
        n++;
    }
    req->number = n;

    return 0;
} // end function set_more_request

/**
 * @brief       put job submit option to environment variables,
 *              and run usub to change it if needed.
 *
 * @param[out]  jf           #1: job file data
 * @param[in]   req          #2: submit/modify request key values
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed
 *
 * @note        environment variables will be put into job file data.
 */
int
setenv_for_submitext(struct lenData *jf, struct request *req)
{
#define QP_SUBMIT_COMMANDNAME 0
    struct config_param myParams[] = {
        {"QP_SUBMIT_COMMANDNAME", STRING_PARAM, -1, {NULL}},
        {NULL, STRING_PARAM, -1, {NULL}}
    };

    char strSubVarFile[PATH_MAX+1];
    snprintf(strSubVarFile, PATH_MAX, "%s/.subparm.%d", gstrTmpDir, (int)getpid());

    FILE *fpSubVar;
    fpSubVar = fopen(strSubVarFile, "w");
    if (NULL == fpSubVar) {
        quickpool_errno = ERROR_SYS_CALL;
        return -1;
    }

    int fd = fileno(fpSubVar);
    fchmod(fd, 0666);

    unsigned int i;
    unsigned int options=0, options2=0, deleteOptions=0, deleteOptions2=0;
    for (i=0; i<req->number; i++) {
        switch (req->keyvalues[i].key) {
        case REQUEST_KEY_SUBMIT_OPTIONS:
            options = atoi(req->keyvalues[i].value);
            break;
        case REQUEST_KEY_SUBMIT_OPTIONS2:
            options2 = atoi(req->keyvalues[i].value);
            break;
        case REQUEST_KEY_MODIFY_DELETE_OPTIONS:
            deleteOptions = atoi(req->keyvalues[i].value);
            break;
        case REQUEST_KEY_MODIFY_DELETE_OPTIONS2:
            deleteOptions2 = atoi(req->keyvalues[i].value);
            break;
        }
    }

    char *command = NULL;
    enum submitType submit = CMD_SUBMIT_JOB;
    unsigned int number=0, maxxf=0;
    for (i=0; i<req->number; i++) {
        switch (req->keyvalues[i].key) {
        case REQUEST_KEY_SUBMIT_TYPE:
            submit = (enum submitType)atoi(req->keyvalues[i].value);
            if (submit == CMD_MODIFY_JOB) {
                fprintf(fpSubVar, "QP_SUBMIT_MODIFY=Y\n");
            }
            break;
        case REQUEST_KEY_SUBMIT_JOBNAME:
            set_string_variable(fpSubVar, "QP_SUBMIT_JOB_NAME", req->keyvalues[i].value,
                                SUBOPT_JOB_NAME, deleteOptions);
            break;
        case REQUEST_KEY_SUBMIT_QUEUE:
            set_string_variable(fpSubVar, "QP_SUBMIT_QUEUE", req->keyvalues[i].value,
                                SUBOPT_QUEUE, deleteOptions);
            break;
        case REQUEST_KEY_SUBMIT_INFILE:
            set_string_variable(fpSubVar, "QP_SUBMIT_IN_FILE", req->keyvalues[i].value,
                                SUBOPT_IN_FILE, deleteOptions);
            break;
        case REQUEST_KEY_SUBMIT_OUTFILE:
            set_string_variable(fpSubVar, "QP_SUBMIT_OUT_FILE", req->keyvalues[i].value,
                                SUBOPT_OUT_FILE, deleteOptions);
            break;
        case REQUEST_KEY_SUBMIT_ERRFILE:
            set_string_variable(fpSubVar, "QP_SUBMIT_ERR_FILE", req->keyvalues[i].value,
                                SUBOPT_ERR_FILE, deleteOptions);
            break;
        case REQUEST_KEY_SUBMIT_SIGNAL:
            set_string_variable(fpSubVar, "QP_SUBMIT_WINDOW_SIG", req->keyvalues[i].value,
                                SUBOPT_WINDOW_SIG, deleteOptions);
            break;
        case REQUEST_KEY_SUBMIT_DEPENDENCY:
            set_string_variable(fpSubVar, "QP_SUBMIT_DEPEND_COND", req->keyvalues[i].value,
                                SUBOPT_DEPEND_COND, deleteOptions);
            break;
        case REQUEST_KEY_SUBMIT_RESREQ:
            set_string_variable(fpSubVar, "QP_SUBMIT_RES_REQ", req->keyvalues[i].value,
                                SUBOPT_RES_REQ, deleteOptions);
            break;
        case REQUEST_KEY_SUBMIT_PRECMD:
            set_string_variable(fpSubVar, "QP_SUBMIT_PRE_EXEC", req->keyvalues[i].value,
                                SUBOPT_PRE_EXEC, deleteOptions);
            break;
        case REQUEST_KEY_SUBMIT_POSTCMD:
            set_string_variable(fpSubVar, "QP_SUBMIT_POST_EXEC", req->keyvalues[i].value,
                                SUBOPT_POST_EXEC, deleteOptions);
            break;            
        case REQUEST_KEY_SUBMIT_LOGINSHELL:
            set_string_variable(fpSubVar, "QP_SUBMIT_LOGIN_SHELL", req->keyvalues[i].value,
                                SUBOPT_LOGIN_SHELL, deleteOptions);
            break;
        case REQUEST_KEY_SUBMIT_MAIL:
            set_string_variable(fpSubVar, "QP_SUBMIT_MAIL_USER", req->keyvalues[i].value,
                                SUBOPT_MAIL_USER, deleteOptions);
            break;
        case REQUEST_KEY_SUBMIT_PROJECT:
            set_string_variable(fpSubVar, "QP_SUBMIT_PROJECT_NAME", req->keyvalues[i].value,
                                SUBOPT_PROJECT_NAME, deleteOptions);
            break;
        case REQUEST_KEY_SUBMIT_USERGROUP:
            set_string_variable(fpSubVar, "QP_SUBMIT_USER_GROUP", req->keyvalues[i].value,
                                SUBOPT_USER_GROUP, deleteOptions);
            break;
        case REQUEST_KEY_SUBMIT_JOBPRIORITY:
            set_string_variable(fpSubVar, "QP_SUBMIT_JOB_PRIORITY", req->keyvalues[i].value,
                                SUBOPT_JOB_PRIORITY, deleteOptions);
            break;
        case REQUEST_KEY_SUBMIT_NODE:
            set_string_variable(fpSubVar, "QP_SUBMIT_HOSTS", req->keyvalues[i].value,
                                SUBOPT_HOST, deleteOptions);
            break;
        case REQUEST_KEY_SUBMIT_NUM_PROCESSOR:
            set_number_variable(fpSubVar, "QP_SUBMIT_NUM_PROCESSORS", req->keyvalues[i].value,
                                DEL_NUMPRO);
            break;
        case REQUEST_KEY_SUBMIT_MAX_PROCESSOR:
            set_number_variable(fpSubVar, "QP_SUBMIT_MAX_PROCESSORS", req->keyvalues[i].value,
                                DEL_NUMPRO);
            break;
        case REQUEST_KEY_SUBMIT_BEGINTIME:
            set_number_variable(fpSubVar, "QP_SUBMIT_BEGIN_TIME", req->keyvalues[i].value,
                                DELETE_NUMBER);
            break;
        case REQUEST_KEY_SUBMIT_TERMTIME:
            set_number_variable(fpSubVar, "QP_SUBMIT_TERM_TIME", req->keyvalues[i].value,
                                DELETE_NUMBER);
            break;
        case REQUEST_KEY_SUBMIT_CPULIMIT:
            set_number_variable(fpSubVar, "QP_SUBMIT_RLIMIT_CPU", req->keyvalues[i].value,
                                DELETE_NUMBER);
            break;
        case REQUEST_KEY_SUBMIT_FILELIMIT:
            set_number_variable(fpSubVar, "QP_SUBMIT_RLIMIT_FSIZE", req->keyvalues[i].value,
                                DELETE_NUMBER);
            break;
        case REQUEST_KEY_SUBMIT_DATALIMIT:
            set_number_variable(fpSubVar, "QP_SUBMIT_RLIMIT_DATA", req->keyvalues[i].value,
                                DELETE_NUMBER);
            break;
        case REQUEST_KEY_SUBMIT_STACKLIMIT:
            set_number_variable(fpSubVar, "QP_SUBMIT_RLIMIT_STACK", req->keyvalues[i].value,
                                DELETE_NUMBER);
            break;
        case REQUEST_KEY_SUBMIT_CORELIMIT:
            set_number_variable(fpSubVar, "QP_SUBMIT_RLIMIT_CORE", req->keyvalues[i].value,
                                DELETE_NUMBER);
            break;
        case REQUEST_KEY_SUBMIT_RSSLIMIT:
            set_number_variable(fpSubVar, "QP_SUBMIT_RLIMIT_RSS", req->keyvalues[i].value,
                                DELETE_NUMBER);
            break;
        case REQUEST_KEY_SUBMIT_SWAPLIMIT:
            set_number_variable(fpSubVar, "QP_SUBMIT_RLIMIT_SWAP", req->keyvalues[i].value,
                                DELETE_NUMBER);
            break;
        case REQUEST_KEY_SUBMIT_RUNLIMIT:
            set_number_variable(fpSubVar, "QP_SUBMIT_RLIMIT_RUN", req->keyvalues[i].value,
                                DELETE_NUMBER);
            break;
        case REQUEST_KEY_SUBMIT_PROCESSLIMIT:
            set_number_variable(fpSubVar, "QP_SUBMIT_RLIMIT_RPOCESS", req->keyvalues[i].value,
                                DELETE_NUMBER);
            break;
        case REQUEST_KEY_SUBMIT_NOFILELIMIT:
            set_number_variable(fpSubVar, "QP_SUBMIT_RLIMIT_NOFILE", req->keyvalues[i].value,
                                DELETE_NUMBER);
            break;
        case REQUEST_KEY_SUBMIT_OPENMAXLIMIT:
            set_number_variable(fpSubVar, "QP_SUBMIT_RLIMIT_OPEN_MAX", req->keyvalues[i].value,
                                DELETE_NUMBER);
            break;
        case REQUEST_KEY_SUBMIT_NUM_XFILE:
            if (deleteOptions & SUBOPT_OTHER_FILES) {
                fprintf(fpSubVar, "QP_SUBMIT_OTHER_FILES=SUB_RESET\n");
            } else {
                fprintf(fpSubVar, "QP_SUBMIT_OTHER_FILES=%s\n", req->keyvalues[i].value);
                maxxf = atoi(req->keyvalues[i].value);
            }
            break;
        case REQUEST_KEY_SUBMIT_XFILE:
            if (number >= maxxf) {
                break;
            }

            fprintf(fpSubVar, "QP_SUBMIT_OTHER_FILES_%d=\"%s\"\n", number, req->keyvalues[i].value);
            number ++;

            break;
        case REQUEST_KEY_MODIFY_COMMAND:
        case REQUEST_KEY_SUBMIT_COMMAND:
            command = req->keyvalues[i].value;
            break;
        }
    }    

    set_option_variable(fpSubVar, "QP_SUBMIT_NOTIFY_BEGIN", SUBOPT_NOTIFY_BEGIN,
                        options, deleteOptions);
    set_option_variable(fpSubVar, "QP_SUBMIT_NOTIFY_END", SUBOPT_NOTIFY_END, options, deleteOptions);
    set_option_variable(fpSubVar, "QP_SUBMIT_EXCLUSIVE", SUBOPT_EXCLUSIVE, options, deleteOptions);
    set_option_variable(fpSubVar, "QP_SUBMIT_RERUNNABLE", SUBOPT_RERUNNABLE, options, deleteOptions);
    set_option_variable(fpSubVar, "QP_SUBMIT_HOLD", SUBOPT2_HOLD, options2, deleteOptions2);

    if (init_config_param(myParams, NULL) < 0) {
        return -1;
    }

    if (NULL != myParams[QP_SUBMIT_COMMANDNAME].string_value) {

        unsigned int lenCmd;
        lenCmd = strlen(command);

        unsigned int lenCmdNoOpt=0;
        while (lenCmdNoOpt < lenCmd) {

            if (isspace(command[lenCmdNoOpt])) {
                break;
            }

            lenCmdNoOpt ++;
        }

        fprintf(fpSubVar, "%s=\"", "QP_SUBMIT_COMMANDNAME");

        for (i=0; i<lenCmdNoOpt; i++) {

            if ('"' == command[i] || '\\' == command[i]) {
                fprintf(fpSubVar, "\\");
            }

            fprintf(fpSubVar, "%c", command[i]);
        }
        fprintf(fpSubVar, "\"\n");

        char *strCmd = unwrap_command(command);
        fprintf(fpSubVar,"QP_SUBMIT_COMMAND_LINE=\"%s\"\n", strCmd);
        FREEUP(strCmd);
    }

    fclose(fpSubVar);

    gf_setenv("QP_SUBMIT_ABORT", (char *)"97");
    gf_setenv("QP_SUBMIT_PARM_FILE", strSubVarFile);

    jf->len = 0;
    jf->data = NULL;

    char *myargv[6];
    char esub[PATH_MAX+1];

    myargv[0] = esub;
    snprintf(esub, PATH_MAX, "%s/%s", baseParams[BASE_SERVERDIR].string_value, ESUBNAME);
    myargv[1] = NULL;
    myargv[1] = (char *)"-r";
    myargv[2] = NULL;

    struct stat sbuf;
    if (0 > stat(esub, &sbuf)) { // no usub script.

        unlink(strSubVarFile);
        if (errno == ENOENT) {
            return 0;
        }

        log_message(ERR, NOCLASS, STR_FUNC_S_FAIL_M, __func__, "stat", esub);

        return -1;
    }

    int ret;
    ret = run_pipe(jf, myargv, getuid());
    if (0 > ret) {

        unlink(strSubVarFile);

        char *strDeltaFile;
        strDeltaFile = getenv("QP_SUBMIT_MODIFY_FILE");
        if (NULL != strDeltaFile) {

            struct stat st;
            if (0 > stat(strDeltaFile, &st) && errno != ENOENT) {
                unlink(strDeltaFile);
            }
        }

        strDeltaFile = getenv("QP_SUBMIT_MODIFY_ENVFILE");
        if (NULL != strDeltaFile) {

            struct stat st;
            if (0 > stat(strDeltaFile, &st) && errno !=ENOENT) {
                unlink(strDeltaFile);
            }
        }

        return -1;
    }

    unlink(strSubVarFile);

    return 0;
} // end function setenv_for_submitext

/* ------------------------------------------------
 *
 *  below are static functions used in this file
 *
 * ------------------------------------------------ */

/**
 * @brief       send submit request to the cluster and get reply from it.
 *
 * @param[in]   submitreq    #1: job submit request key values
 * @param[in]   jf           #2: job file data
 * @param[out]  submitreply  #3: submit job reply from JS
 * @param[in]   master       #4: master(port and name) to be called
 *
 * @retval      > 0          #1: succeed, got job ID from JS
 * @retval      -1           #2: failed
 *
 * @note        caller should calloc memory for QuickPoolSubmitReply if need.
 *              For JS replied error, caller should handle it.
 */
static JOBID_T
send_batch(struct request *submitreq, struct lenData *jf,
           struct QuickPoolSubmitReply *submitreply, struct poolMaster *master)
{
    struct auth auth;
    gf_set_auth(&auth, commandParams[AUTH_KIND].string_value);

    unsigned int size; // for calculating the size of the request.
    size = gf_xdrsize_request(submitreq);
    size += gf_xdrsize_auth(&auth);
    size += HEADER_LEN;

    size += jf->len + sizeof(int);

    char *request;
    request = (char *)calloc(size, sizeof(char));
    if (NULL == request) {

        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", size);
        gf_xdr_free(gf_xdr_auth, (void *)&auth);
        quickpool_errno = ERROR_MEMORY;
        return -1;
    }

    struct header tohdr, outhdr;
    gf_init_header(&tohdr);
    tohdr.reqcode = BATCH_JOB_SUB;

    XDR xdrs;
    xdrmem_create(&xdrs, request, size, XDR_ENCODE);

    if (!gf_xdr_message(&xdrs, submitreq, &tohdr, &auth, gf_xdr_request, NULL)) {

        // fail to encode request.
        log_message(ERR, NOCLASS, STR_FUNC_FAIL, __func__, "gf_xdr_message/gf_xdr_request");
        xdr_destroy(&xdrs);
        FREEUP(request);
        gf_xdr_free(gf_xdr_auth, (void *)&auth);

        return -1;
    }

    //put job file len and data after submitreq buffer
    int pos, nlen;
    if (0 < jf->len && NULL != jf->data) {
        nlen = htonl(jf->len);
        pos = XDR_GETPOS(&xdrs);
        memcpy(request + pos, (char*)&nlen, sizeof(int));
        memcpy(request + pos + sizeof(int), jf->data, jf->len);
        pos += jf->len + sizeof(int);
        XDR_SETPOS(&xdrs, pos);
    }

    char *masterList;
    unsigned int port;
    unsigned int connTimeout;
    unsigned int recvTimeout;
    if (NULL != master) { // client set master, let's call it directly.
        masterList = master->master;
        port = master->port;
        connTimeout = master->conntimeout;
        recvTimeout = master->recvtimeout;
    } else {
        masterList = qp_getmastername();
        if (NULL == masterList) {
            // JM do not know who is master, let's try g_masterlist.
            masterList = g_masterlist;
        }

        port = g_js_port;
        connTimeout = g_conntimeout;
        recvTimeout = g_recvtimeout;
    }

    char *reply;
    int cc, fd;
    cc = gf_callmaster(masterList, port, connTimeout, recvTimeout, g_connretries,
                       XDR_GETPOS(&xdrs), request, &outhdr, &reply, &fd);

    if (0 < fd) {
        gf_chan_close(fd);
    }
    xdr_destroy(&xdrs);
    FREEUP(request);
    gf_xdr_free(gf_xdr_auth, (void *)&auth);

    if (0 >  cc) {

        FREEUP(reply);
        log_message(ERR, NOCLASS, STR_FUNC_S_FAIL, __func__, "gf_callmaster", masterList);
        return -1;
    }

    // got job submit reply from JS, decode it.
    xdrmem_create(&xdrs, reply, outhdr.length, XDR_DECODE);
    if (!libxdr_submit_reply(&xdrs, submitreply, &outhdr, NULL)) {

        log_message(ERR, NOCLASS, STR_FUNC_FAIL, __func__, "libxdr_submit_reply");
        xdr_destroy(&xdrs);
        FREEUP(reply);

        return -1;
    }

    xdr_destroy(&xdrs);
    FREEUP(reply);

    quickpool_errno = outhdr.reqcode;
    if (quickpool_errno == 0) {
        return (submitreply->jobid);
    }

    return -1;
} // end function send_batch

/**
 * @brief       get chkpnt path from the given dir string.
 *
 * @param[in]   dir          #1: give chkpnt dir
 * @param[out]  path         #2: chkpnt path
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed
 */
static int
get_chkpnt_path(char *dir, char *path)
{
    if (NULL == dir) {
        log_message(ERR, NOCLASS, "%s: input dir is NULL.", __func__);
        return -1;
    }

    char *sp;
    sp = strrchr(dir, '/');
    if (NULL != sp && islong(sp+1)) {
        snprintf(path, PATH_MAX, "%s", dir);
        return 0;
    }

    DIR *dirEnt;
    dirEnt = opendir(dir);
    if (NULL == dirEnt) {
        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "opendir" );
        return -1;
    }

    char strJobIdDir[PATH_MAX];
    strJobIdDir[0] = '\0';

    struct dirent *dirContent;
    while (NULL != (dirContent=readdir(dirEnt))) {

        if (islong(dirContent->d_name)) {
            strcpy(strJobIdDir, dirContent->d_name);
            break;
        }
    }

    closedir(dirEnt);

    if (0 >= strlen(strJobIdDir)) {
        log_message(ERR, NOCLASS, "%s: not find a job chknpt directory.", __func__);
        return -1;
    }

    snprintf(path, PATH_MAX, "%s/%s", dir, strJobIdDir);

    return 0;
} // end function get_chkpnt_path

/**
 * @brief       send restart request to the cluster and get reply from it.
 *
 * @param[in]   req          #1: job restart request key values
 * @param[out]  submitreply  #2: submit job reply from JS
 * @param[in]   master       #3: master(port and name) to be called
 *
 * @retval      > 0          #1: succeed, got job ID from JS
 * @retval      -1           #2: failed
 *
 * @note        caller should calloc memory for QuickPoolSubmitReply if need.
 *              For JS replied error, caller should handle it.
 */
static JOBID_T
send_restart(struct request *req, struct QuickPoolSubmitReply *submitrep,
             struct poolMaster *master)
{
    char *project=NULL, *command=NULL;
    STATUS_WAIT_T status;
    struct {
        int error;
        int eno;
    } err;

    char chkPath[PATH_MAX+1];
    memset(chkPath, 0, PATH_MAX);
    err.error = err.eno = 0;

    int childIoFd[2];
    if (socketpair(AF_UNIX, SOCK_STREAM, 0, childIoFd) < 0) {
        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "socketpair");
        return -1;
    }

    unlink_delta_files();

    pid_t pid;
    pid = fork();
    if (0 > pid) {
        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "fork");
        return -1;
    } else if (0 == pid) {
        char chklog[PATH_MAX];
        FILE *fp=NULL;
        int exitVal = -1;

        close (childIoFd[0]);

        struct linger linstr = {1, 5};
        setsockopt(childIoFd[1], SOL_SOCKET, SO_LINGER, (char *)&linstr,
                   sizeof(linstr));

        struct record *rec=NULL;
        char *strChkpntDir=NULL;

        uid_t uid;
        uid = getuid();
        if (0 > setuid(uid)) {
            log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "setuid");
            goto sendErr;
        }

        unsigned int i;
        for (i=0; i<req->number; i++) {
            switch (req->keyvalues[i].key) {
            case REQUEST_KEY_RESTART_CHKPNTDIR:
                strChkpntDir = req->keyvalues[i].value;
                break;
            }
        }

        int cc;
        cc = get_chkpnt_path(strChkpntDir, chkPath);
        if (0 > cc) {
            log_message(ERR, NOCLASS, STR_FUNC_FAIL, __func__, "get_chkpnt_path");
            goto sendErr;
        } else {
            snprintf(chklog, PATH_MAX, "%s/chklog", chkPath);
        }

        if (write(childIoFd[1], (char *) chkPath, sizeof(chkPath)) != sizeof(chkPath)) {
            log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "write");
            goto childExit;
        }

        if (cc == -1) goto sendErr;

        if ((fp = fopen(chklog, "r")) == NULL) {
            log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "fopen");
            goto sendErr;
        }

        unsigned int numLine;
        rec = read_file_line(fp, &numLine);
        if (NULL == rec || rec->type != RECORD_TYPE_JOB_NEW) {

            log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "read_file_line");
            fclose(fp);
            goto sendErr;
        }

        err.error = FALSE;
        if (write(childIoFd[1], (char *) &err, sizeof(err)) != sizeof(err)) {
            log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "write" );
            goto childExit;
        }

        unsigned int writeLen;
        writeLen = gf_chan_write(childIoFd[1], (char *)&(rec->numKVs), sizeof(unsigned int), 10);
        if (writeLen != sizeof(unsigned int)) {
            log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "gf_chan_write(numKVs)");
            goto childExit;
        }

        for (i=0; i<rec->numKVs; i++) {
            /* todo: allow empty value so far, revisit later
            if (strlen(rec->KVs[i].value) <= 0) {
                log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "value is null");
                goto childExit;
            }
            */

            writeLen = gf_chan_write(childIoFd[1], (char *)&(rec->KVs[i].key),
                                     sizeof(unsigned int), 10);
            if (writeLen != sizeof(unsigned int)) {
                log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "gf_chan_write(key)");
                goto childExit;
            }

            unsigned int length;
            length = strlen(rec->KVs[i].value);
            writeLen = gf_chan_write(childIoFd[1], (char *)&(length), sizeof(unsigned int), 10);
            if (writeLen != sizeof(unsigned int)) {
                log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "gf_chan_write(value len)");
                goto childExit;
            }

            writeLen = gf_chan_write(childIoFd[1], rec->KVs[i].value, length, 10);
            if (writeLen != length) {
                log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "gf_chan_write(value string");
                goto childExit;
            }
        }
      
        exitVal = 0;

    childExit:

        if (NULL != fp) {
            fclose(fp);
            fp = NULL;
        }

        close(childIoFd[1]);
        exit(exitVal);

    sendErr:
        err.error = TRUE;
        err.eno = errno;

        if (NULL != rec) {
            free_record(rec);
            FREEUP(rec);
        }

        if (write(childIoFd[1], (char *) &err, sizeof(err)) != sizeof(err)) {
            close(childIoFd[1]);
            exit(-1);
        }
        close(childIoFd[1]);
        exit(0);
    }

    close (childIoFd[1]);

    unsigned int options=0, options2=0;
    unsigned int recOptions=0, recOptions2=0;
    int opt2Index = 0;

    if (read(childIoFd[0], chkPath, sizeof(chkPath)) != sizeof(chkPath)) {
        goto parentErr;
    }

    err.error = FALSE;
    if (read(childIoFd[0], (char *) &err, sizeof(err)) != sizeof(err)) {
        goto parentErr;
    }
    if (err.error) {
        errno = err.eno;
        goto parentErr;
    }

    unsigned int readLen, numKVs;
    readLen = gf_chan_read(childIoFd[0], (char *)&numKVs, sizeof(unsigned int), 10);
    if (readLen != sizeof(unsigned int)) {
        goto parentErr;
    }

    unsigned int i;
    for (i=0; i<req->number; i++) {
        switch (req->keyvalues[i].key) {
        case REQUEST_KEY_SUBMIT_OPTIONS:
            options = atoi(req->keyvalues[i].value);
            break;
        case REQUEST_KEY_SUBMIT_OPTIONS2:
            options2 = atoi(req->keyvalues[i].value);
            opt2Index = i;
            break;
        }
    }

    for (i=0; i<numKVs; i++) {
        unsigned int key;
        readLen = gf_chan_read(childIoFd[0], (char *)&key, sizeof(unsigned int), 10);
        if (readLen != sizeof(unsigned int)) {
            goto parentErr;
        }

        unsigned int valueLen = 0;
        readLen = gf_chan_read(childIoFd[0], (char *)&valueLen, sizeof(unsigned int), 10);
        if (readLen != sizeof(unsigned int)) {
            goto parentErr;
        }

        /* todo: allow empty value so far, revisit later
        if (valueLen <= 0) {
            goto parentErr;
        }
        */
        char *value = (char *)calloc(valueLen, sizeof(char));
        if (NULL == value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", valueLen);
            goto parentErr;
        }

        readLen = gf_chan_read(childIoFd[0], value, valueLen, 10);
        if (readLen != valueLen) {
            goto parentErr;
        }

        int sigValue;
        switch (key) {        
        case RECORD_KEY_USERID:
            set_restart_number_request(req, REQUEST_KEY_SUBMIT_USERID, value);
            FREEUP(value);

            break;           
        case RECORD_KEY_OPTIONS:            
            recOptions = atoi(value);   
            if (recOptions & SUBOPT_RERUNNABLE) {
                options |= SUBOPT_RERUNNABLE;
            }
            FREEUP(value);

            break;                    
        case RECORD_KEY_OPTIONS2:            
            recOptions2 = atoi(value);  
            if (recOptions2 & SUBOPT2_HOLD) {
                options2 |= SUBOPT2_HOLD;
            }
            FREEUP(value);

            break;            
        case RECORD_KEY_PROCESSOR:
            set_restart_number_request(req, REQUEST_KEY_SUBMIT_NUM_PROCESSOR, value);
            options2 &= ~SUBOPT2_DEF_PROCLIMIT;
            FREEUP(value);

            break;
        case RECORD_KEY_MAX_PROCESSOR:
            set_restart_number_request(req, REQUEST_KEY_SUBMIT_MAX_PROCESSOR, value);
            FREEUP(value);

            break;  
        case RECORD_KEY_SCHEDULE_HOSTTYPE:
            set_restart_string_request(req, RECORD_KEY_SCHEDULE_HOSTTYPE, value);

            break;
        case RECORD_KEY_LOGIN_SHELL:
            set_restart_string_request(req, REQUEST_KEY_SUBMIT_LOGINSHELL, value);
            FREEUP(value);

            break;   
        case RECORD_KEY_SIGNAL:
            sigValue = atoi(value);
            if (sigValue > _NSIG || sigValue < 0) {
                goto parentErr;
            }    

            set_restart_number_request(req, REQUEST_KEY_SUBMIT_SIGNAL, value);
            FREEUP(value);

            break;
        case RECORD_KEY_UMASK:
            set_restart_number_request(req, REQUEST_KEY_SUBMIT_UMASK, value);
            FREEUP(value);

            break;
        case RECORD_KEY_QUEUE:
            set_restart_string_request(req, REQUEST_KEY_SUBMIT_QUEUE, value);
            FREEUP(value);

            break;
        case RECORD_KEY_RESREQ:
            set_restart_string_request(req, REQUEST_KEY_SUBMIT_RESREQ, value);
            FREEUP(value);

            break;
        case RECORD_KEY_CWD:
            set_restart_string_request(req, REQUEST_KEY_SUBMIT_CWD, value);
            FREEUP(value);

            break;
        case RECORD_KEY_SUBMIT_HOMEDIR:
            set_restart_string_request(req, REQUEST_KEY_SUBMIT_HOMEDIR, value);
            FREEUP(value);

            break;
        case RECORD_KEY_SUBMIT_INFILE:
            set_restart_string_request(req, REQUEST_KEY_SUBMIT_INFILE, value);
            FREEUP(value);

            break;   
        case RECORD_KEY_SUBMIT_OUTFILE:
            set_restart_string_request(req, REQUEST_KEY_SUBMIT_OUTFILE, value);
            FREEUP(value);

            break;   
        case RECORD_KEY_SUBMIT_ERRFILE:
            set_restart_string_request(req, REQUEST_KEY_SUBMIT_ERRFILE, value);
            FREEUP(value);

            break;
        case RECORD_KEY_JOB_FILE:
            set_restart_string_request(req, REQUEST_KEY_RESTART_JOBFILE, value);
            FREEUP(value);

            break;  
        case RECORD_KEY_ASKEDHOST:
            set_restart_string_request(req, REQUEST_KEY_SUBMIT_NODE, value);
            FREEUP(value);

            break;            
        case RECORD_KEY_JOB_NAME:
            set_restart_string_request(req, REQUEST_KEY_SUBMIT_JOBNAME, value);
            FREEUP(value);

            break;
        case RECORD_KEY_COMMAND:
            command = value;
            set_restart_string_request(req, REQUEST_KEY_SUBMIT_COMMAND, value);

            break;   
        case RECORD_KEY_MAIL_USER:
            set_restart_string_request(req, REQUEST_KEY_SUBMIT_MAIL, value);
            FREEUP(value);

            break;
        case RECORD_KEY_PROJECT_NAME:
            project = value;
            set_restart_string_request(req, REQUEST_KEY_SUBMIT_PROJECT, value);

            break;  
        case RECORD_KEY_NUM_XFILE:
            set_restart_number_request(req, REQUEST_KEY_SUBMIT_NUM_XFILE, value);
            FREEUP(value);

            break;
        case RECORD_KEY_XFILE:
            set_restart_string_request(req, REQUEST_KEY_SUBMIT_XFILE, value);
            FREEUP(value);

            break;
        case RECORD_KEY_USER_PRIORITY:
            set_restart_number_request(req, REQUEST_KEY_SUBMIT_JOBPRIORITY, value);
            FREEUP(value);

            break;  
        case RECORD_KEY_CHECKPOINT_METHOD:
            set_restart_number_request(req, REQUEST_KEY_CHKPNT_METHOD, value);
            FREEUP(value);

            break;
        case RECORD_KEY_CHECKPOINT_PERIOD:
            set_restart_number_request(req, REQUEST_KEY_CHKPNT_PERIOD, value);
            FREEUP(value);

            break;   
        case RECORD_KEY_JOB_PGID:
            set_restart_number_request(req, REQUEST_KEY_RESTART_JOBPID, value);
            FREEUP(value);
            break;
        default :  
            FREEUP(value);
            log_message(DEBUG, NOCLASS, "%s do not need the key : %d", __func__, key);
            break;
        }
    }

    close(childIoFd[0]);
 
    if (opt2Index) {
        snprintf(req->keyvalues[opt2Index].value, 12, "%d", options2);
    }

    if (waitpid(pid, &status, 0) < 0) {
        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "waitpid" );
        return -1;
    }

    struct lenData jf;
    if (0 > set_jobfile_data(project, command, &jf)) {
        FREEUP(project);
        FREEUP(command);
        return -1;
    }
    FREEUP(project);
    FREEUP(command);

    JOBID_T jobId;
    jobId = send_batch(req, &jf, submitrep, master);
    
    return (jobId);

parentErr:
    if (!err.error) {
        err.eno = errno;
    }
    close(childIoFd[0]);
    if (waitpid (pid, &status, 0) < 0) {
        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "waitpid" );
    } else {
        errno = err.eno;
    }

    return -1;
} // end function send_restart

#define QUOTE_STR(_str1,_str)  {                                        \
        int i, j, cnt=0;                                                \
        char ch, next, *tmp_str=NULL;                                   \
        for (i=0; _str1[i]; i++){                                       \
            if (_str1[i] == '"')                                        \
                cnt ++;                                                 \
        }                                                               \
        tmp_str = (char *)calloc((strlen(_str1)+1+cnt+2), sizeof(char)); \
        if (tmp_str != NULL) {                                          \
            tmp_str[0] = '"';                                           \
            _str = tmp_str+1;                                           \
            strcpy(_str, _str1);                                        \
            for (i=0; _str[i]; ) {                                      \
                if (_str[i] == '"') {                                   \
                    ch = _str[i];                                       \
                    _str[i] = '\\';                                     \
                    next = ch;                                          \
                    for ( j=i+1; _str[j]; j++){                         \
                        ch = _str[j];                                   \
                        _str[j] = next;                                 \
                        next = ch;                                      \
                    }                                                   \
                    _str[j] = next;                                     \
                    _str[j+1] = '\0';                                   \
                    i += 2;                                             \
                }                                                       \
                i++;                                                    \
            }                                                           \
            for (i=0; _str[i]; i++ ){}                                  \
            _str[i] = '"';                                              \
            _str[i+1] = '\0';                                           \
        }                                                               \
        _str = tmp_str;                                                 \
    }


#define MSG_BAD_INTVAL3s ("%s: Bad value(%s) read from "    \
                                      "$QP_SUBMIT_MODIFY_FILE for parameter " \
                                      "%s, the setting will be ignored. " \
                                      "The value of this parameter can " \
                                      "only be SUB_RESET or an integer.")

#define MSG_ALLOC_MEMF_IN_XFs ("%s: Memory allocate failed for " \
                                       "file transfer request.")

#define MSG_BAD_XF_OP2s ("%s: unknown file transfer operator" \
                                     " %s, this transfer request will be" \
                                     " ignored.")

#define MSG_BAD_BOOLVAL3s ("%s: Bad value(%s) read from "    \
                                      "$QP_SUBMIT_MODIFY_FILE for parameter" \
                                      " %s, the setting will be ignored. " \
                                      "The value of this parameter can " \
                                      "only be SUB_RESET or 'Y'.")
  
#define MSG_VAL_RESET_INT2s (\
                                       "%s: The value of %s can only be " \
                                       "SUB_RESET or an integer.")

#define MSG_BAD_ENVAL3s (\
                                     "%s: Bad value(%s) read from "     \
                                     "$QP_SUBMIT_MODIFY_FILE for parameter" \
                                     " %s, the setting will be ignored.")

#define MSG_WARN_NULLVAL2s (\
                                     "%s: The value of parameter %s is " \
                                     "empty, the setting will be ignored.")

void
unlink_delta_files(void)
{
    char parmDeltaFile[PATH_MAX+1];
    char envDeltaFile[PATH_MAX+1];
    struct stat stbuf;

    snprintf(parmDeltaFile, PATH_MAX, "%s/.submit_delta_parameter.%d.%d",
             gstrTmpDir, getpid(), getuid());
    snprintf(envDeltaFile, PATH_MAX, "%s/.submit_delta_env.%d.%d",
             gstrTmpDir, getpid(), getuid());

    if (stat(parmDeltaFile,&stbuf) != ENOENT) {
        unlink(parmDeltaFile);
    }

    if (stat(envDeltaFile,&stbuf) != ENOENT) {
        unlink(envDeltaFile);
    }

    gf_setenv("QP_SUBMIT_MODIFY_FILE", parmDeltaFile);
    gf_setenv("QP_SUBMIT_MODIFY_ENVFILE", envDeltaFile);

    return;
}

static char *
get_user_map(int *fail, char *username)
{
    struct passwd *pw;
    pw = get_user_byname(username);
    if (NULL == pw) {
        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "get_user_byname");
        return NULL;
    }

    char hostfn[PATH_MAX];
    strcpy(hostfn, pw->pw_dir);
    strcat(hostfn,"/.usermap");

    FILE *fileMap;
    fileMap=fopen(hostfn, "r");
    if (NULL == fileMap) {

        if (errno != ENOENT) {
            log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "fopen" );
        }

        return NULL;
    }

    struct stat statbuf;
    if ((fstat(fileno(fileMap), &statbuf) < 0)
        || (statbuf.st_uid != 0 && statbuf.st_uid != pw->pw_uid) || (statbuf.st_mode & 066)) {

        fclose(fileMap);
        return NULL;
    }

    int maplen =256;
    char *map = (char *)calloc(maplen, sizeof(char));
    if (NULL == map) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", maplen);
        *fail=1;
        return NULL;
    }

    int len =0;

    char *line;
    unsigned int lineNum = 0;
    char buf[BUFSIZ];
    while (NULL != (line=gf_get_line(fileMap, &lineNum, buf))) {
        char direction[10], clusorhost[MAXNAMELEN], user[MAXNAMELEN];
        int num;

        num = sscanf(line, "%s %s %s", clusorhost, user, direction);
        if (num < 2 || (num ==3 && strcmp(direction, "send") != 0))
            continue;

        if (strcmp(user,"+") == 0) {
            continue;
        }

        len += strlen(clusorhost) + 1 + strlen(user) + 1;
        if (len > maplen) {
            char *newmap;

            maplen += 256;
            if ((newmap=(char *)realloc(map, maplen)) == NULL) {

                log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "realloc" );

                fclose(fileMap);
                free(map);
                *fail=1;

                return NULL;
            }
            map = newmap;
        }

        strcat(map, clusorhost);
        strcat(map, " ");
        strcat(map, user);
        strcat(map, " ");
    }
    fclose(fileMap);

    return map;
}

/**
 * @brief       fork a child to run usub to change user submit request.
 *
 * @param[in]   req          #1: job submit request key values
 *
 * @retval      0            #1: succeed, usub done successfully
 * @retval      -1           #2: failed
 *
 * @note        use pipe to communicate data between child and parent.
 *              child write error and errno to pipe first,
 *              write usub data len then, write usub data at last.
 */
static int
child_dothings(struct request *req)
{
    int childIoFd[2];
    int len;

    struct {
        int error;
        int eno;
    } err;

    char *qpUserName;
    qpUserName = getmy_uname();

    if (socketpair(AF_UNIX, SOCK_STREAM, 0, childIoFd) < 0) {
        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "socketpair");
        return -1;
    }

    pid_t pid;
    pid = fork();
    if (0 > pid) {
        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "fork");
        return -1;
    } else if (0 < pid) {

        STATUS_WAIT_T status;
        close(childIoFd[1]);

        err.eno = -1;
        if (read(childIoFd[0], (char *)&err, sizeof(err)) != sizeof(err)) {

            log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "read");
            err.error = TRUE;
            goto waitforchild;
        }

        if (err.error) {
            errno = err.eno;
            goto waitforchild;
        }
        err.eno = -1;

        if (read(childIoFd[0], (char *)&(sdataUsub.len), sizeof(int)) != sizeof(int)) {
            err.error = TRUE;
            log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "read");
            goto waitforchild;
        }

        FREEUP(sdataUsub.data);
        if (0 < sdataUsub.len) {

            sdataUsub.data = (char *)calloc(sdataUsub.len+1, sizeof(char));
            if (NULL == sdataUsub.data) {
                err.error = TRUE;
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", sdataUsub.len+1);
                goto waitforchild;
            }

            if (gf_chan_read(childIoFd[0], sdataUsub.data, sdataUsub.len, 10) != sdataUsub.len) {
                FREEUP(sdataUsub.data);
                err.error = TRUE;
                log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "gf_chan_read");
                goto waitforchild;
            }
        } else {
            sdataUsub.data = NULL;
        }

        err.eno = -1;
        if (read(childIoFd[0], (char *) &err, sizeof(err)) != sizeof(err)) {

            log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "read");
            err.error = TRUE;
            goto waitforchild;
        }

        if (err.error) {
            errno = err.eno;
            goto waitforchild;
        }
        err.eno = -1;

        if (read(childIoFd[0], (char *)&len, sizeof(int)) != sizeof(int)) {
            err.error = TRUE;
            log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "read");
            goto waitforchild;
        }

        if (0 < len) {

            FREEUP(sstrUserMap);
            sstrUserMap = (char *)calloc(len, sizeof(char));
            if (NULL == sstrUserMap) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", len);
                err.error = TRUE;
                goto waitforchild;
            }

            if (read(childIoFd[0], sstrUserMap, len) != len) {
                err.error = TRUE;
                log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "read");
                goto waitforchild;
            }
        }

    waitforchild:
        close(childIoFd[0]);
        FREEUP(qpUserName);

        if (0 > err.eno) {
            err.eno = errno;
        }

        if (waitpid(pid, &status, 0) < 0) {
            log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "waitpid");
            return -1;
        }

        if (err.error) {
            errno = err.eno;
            return -1;
        }

        return 0;
    }

    close(childIoFd[0]);

    // set environment variables and run usub in child
    if (setenv_for_submitext(&sdataUsub, req) < 0) {
        goto errorChild;
    } else {
        err.error = FALSE;
        err.eno = -1;
        if (write(childIoFd[1], (char *)&err, sizeof(err)) != sizeof(err)) {
            close(childIoFd[1]);
            exit(-1);
        }
    }

    if (write(childIoFd[1], (char *)&(sdataUsub.len), sizeof(int)) != sizeof(int)) {
        close(childIoFd[1]);
        exit(-1);
    }

    if (0 < sdataUsub.len) {
        if (write(childIoFd[1], (char *)sdataUsub.data, sdataUsub.len) != sdataUsub.len) {
            close(childIoFd[1]);
            exit(-1);
        }
    }

    int fail;
    fail = 0;

    char *strUserMap;
    strUserMap = get_user_map(&fail, qpUserName);
    if (NULL == strUserMap) {
        len = 0;
    } else {
        len = strlen(strUserMap) + 1;
    }

    if (fail) {
        goto errorChild;
    } else {
        err.error = FALSE;
        err.eno = -1;
        if (write(childIoFd[1], (char *) &err, sizeof(err)) != sizeof(err)) {
            close(childIoFd[1]);
            exit(-1);
        }
    }

    if (write(childIoFd[1], (char *)&len, sizeof(int)) != sizeof(int)) {
        close(childIoFd[1]);
        exit(-1);
    }

    if (len > 0) {
        if (write(childIoFd[1], (char *)strUserMap, len) != len) {
            close(childIoFd[1]);
            exit(-1);
        }
    }
    exit(0);

errorChild:

    err.error = TRUE;
    err.eno = errno;

    if (write(childIoFd[1], (char *) &err, sizeof(err)) != sizeof(err)) {
        exit(-1);
    }

    exit(0);
} // end function child_dothings

static void
set_string_variable(FILE *fp, const char *name, char *value,
                    unsigned int option, unsigned int deleteOptions)
{
    char *quoteValue;

    QUOTE_STR(value , quoteValue);
    if (NULL != quoteValue) {
        fprintf(fp, "%s=%s\n", name, quoteValue);
        FREEUP(quoteValue);
    } else if (deleteOptions & option) {
        fprintf(fp, "%s=SUB_RESET\n", name);
    }

    return;
} // end function set_string_variable

static void
set_number_variable(FILE *fp, const char *name, char *value, int deleteValue)
{
    if (deleteValue == atoi(value)) {
        fprintf(fp, "%s=SUB_RESET\n", name);
    } else {
        fprintf(fp, "%s=%s\n", name, value);
    }

    return;
} // end function set_number_variable

static void
set_option_variable(FILE *fp, const char *name, unsigned int option,
                    unsigned int options, unsigned int deleteOptions)
{
    if (options & option) {
        fprintf(fp, "%s=Y\n", name);
    } else if (deleteOptions & option) {
        fprintf(fp, "%s=SUB_RESET\n", name);
    }

    return;
} // end function set_option_variable

static void
set_restart_string_request(struct request *req, unsigned int key, char *value)
{
    unsigned int i;
    for (i=0; i<req->number; i++) {
        if (req->keyvalues[i].key != key) {
            continue;
        }

        FREEUP(req->keyvalues[i].value);
        req->keyvalues[i].value = copy_string(value);

        break;
    }

    if (i == req->number) {
        req->keyvalues[req->number].key = key;
        req->keyvalues[req->number].value = copy_string(value);
        req->number ++;
    }

    return;
}

static void
set_restart_number_request(struct request *req, unsigned int key, char *value)
{
    unsigned int i;
    for (i=0; i<req->number; i++) {
        if (req->keyvalues[i].key != key) {
            continue;
        }

        snprintf(req->keyvalues[i].value, 12, "%d", atoi(value));
        break;
    }

    if (i == req->number) {
        req->keyvalues[req->number].key = key;
        req->keyvalues[req->number].value = copy_string(value);
        req->number ++;
    }

    return;
}

/**
 * @brief       command from script need handle _USER_SCRIPT_ for set
 *              QP_SUBMIT_COMMAND_LINE.
 *
 * @param[in]   command      #1: job command from command line or script
 *
 * @retval      not NULL     #1: real job command
 */
static char *
unwrap_command(char *command)
{
    char *strUnwrapCmd;
    strUnwrapCmd = copy_string(command);

    char *sp;
    sp = strstr(strUnwrapCmd, "SCRIPT_\n");
    if (NULL == sp) {
        return strUnwrapCmd;
    }

    char *strNoScriptCmd;
    strNoScriptCmd = sp+strlen("SCRIPT_\n");

    sp = strstr(strNoScriptCmd,"SCRIPT_\n");
    if (NULL == sp) {
        FREEUP(strUnwrapCmd);
        strUnwrapCmd = copy_string(strNoScriptCmd);
        return strUnwrapCmd;
    }

    while ('\n' != *sp) sp --;

    sp ++;
    *sp = '\0';

    FREEUP(strUnwrapCmd);
    strUnwrapCmd = copy_string(strNoScriptCmd);

    char *start;
    bool_t hasNonSpace=FALSE;

    sp = NULL;
    start = strUnwrapCmd;
    while (*start) {

        if ('\n' == *start) {

            *start = ' ';
            if (NULL != sp) {
                if (hasNonSpace) {
                    *sp = ';';
                } else {
                    *sp = ' ';
                }
            }

            sp = start;
            hasNonSpace = FALSE;
        } else if(!isspace(*start)) {
            hasNonSpace = TRUE;
        }

        start ++;
    }

    return strUnwrapCmd;
} // end function unwrap_command

static void
handle_command_argument(char *line, struct request *req, enum submitType submit, const char *token)
{
    line = strip_spaces(line, FALSE);

    if (0 == strlen(line)) {
        log_message(WARNING, NOCLASS, MSG_WARN_NULLVAL2s, __func__, token);
        return;
    }

    unsigned int key;
    if (CMD_MODIFY_JOB == submit) {
        key = REQUEST_KEY_MODIFY_COMMAND;
    } else {
        key = REQUEST_KEY_SUBMIT_COMMAND;
    }

    int index = -1;
    unsigned int j;
    for (j=0; j<req->number; j++) {

        if (req->keyvalues[j].key != key) {
            continue;
        }

        index = j;
        break;
    }

    char *sp;
    sp = get_string_inside(&line, '"', '"');

    if (NULL != strchr(sp, '\n')) {
        const char *szTmpShellCommands= "\n_USER_SCRIPT_\n) > $QP_CHKFILENAME.shell\n"
                                        "chmod u+x $QP_CHKFILENAME.shell\n"
                                        "$QP_JOBFILENAME.shell\n"
                                        "saveExit=$?\n"
                                        "/bin/rm -f $QP_JOBFILENAME.shell\n"
                                        "(exit $saveExit)\n";

        unsigned int length = strlen(sp) + strlen(szTmpShellCommands) + 200;
        char *cmd = (char *)calloc(length, sizeof(char));
        if (NULL == cmd) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", length);
            return;
        }
        snprintf(cmd, length, "(cat <<_USER_\\SCRIPT_\n%s\n%s", sp, szTmpShellCommands);
        FREEUP(sp);
        sp = cmd;
    }

    if (0 <= index) {
        FREEUP(req->keyvalues[index].value);
        req->keyvalues[index].value = sp;
    } else {
        req->keyvalues[req->number].value = sp;
        req->number ++;
    }

    return;
} // end function handle_command_argument

static void
handle_string_argument(char *line, struct request *req, unsigned int key, unsigned int option,
                       unsigned int *delOptions, const char *token)
{
    line = strip_spaces(line, FALSE);

    if (0 == strlen(line)) {
        log_message(WARNING, NOCLASS, MSG_WARN_NULLVAL2s, __func__, token);
        return;
    }

    if (is_stringtoken(line, "SUB_RESET")) {
        *delOptions |= option;
        return;
    }

    int index = -1;
    unsigned int j;
    for (j=0; j<req->number; j++) {

        if (req->keyvalues[j].key != key) {
            continue;
        }

        index = j;
        break;
    }

    char *sp;
    sp = get_string_inside(&line, '"', '"');

    if (REQUEST_KEY_SUBMIT_SIGNAL == key) {
        int signalValue = gf_string_2signal(sp);
        FREEUP(sp);
        sp = (char *)calloc(12, sizeof(char));
        if (NULL == sp) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            return;
        }
        snprintf(sp, 12, "%d", signalValue);
    }

    if (0 <= index) {
        FREEUP(req->keyvalues[index].value);
        req->keyvalues[index].value = sp;
    } else {
        req->keyvalues[req->number].value = sp;
        req->number ++;
    }

    *delOptions &= ~option;

    return;
} // end function handle_string_argument

static void
handle_number_argument(char *line, struct request *req, unsigned int key, unsigned int option,
                       unsigned int *deleteOptions, const char *token)
{
    line = strip_spaces(line, FALSE);

    if (0 == strlen(line)) {
        log_message(WARNING, NOCLASS, MSG_WARN_NULLVAL2s, __func__, token);
        return;
    }

    int index = -1;
    unsigned int j;
    for (j=0; j<req->number; j++) {

        if (req->keyvalues[j].key != key) {
            continue;
        }

        index = j;
        break;
    }

    char *sp;
    if (is_stringtoken(line, "SUB_RESET")) {

        if (REQUEST_KEY_SUBMIT_JOBPRIORITY == key) {
            *deleteOptions |= option;
            return;
        } else if (REQUEST_KEY_SUBMIT_NUM_PROCESSOR == key
                   || REQUEST_KEY_SUBMIT_MAX_PROCESSOR == key) {
            sp = (char *)calloc(12, sizeof(char));
            if (NULL == sp) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                return;
            }
            snprintf(sp, 12, "%d", DEL_NUMPRO);
        } else {
            sp = (char *)calloc(12, sizeof(char));
            if (NULL == sp) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                return;
            }
            snprintf(sp, 12, "%d", DELETE_NUMBER);
        }
    } else {
        sp = get_string_inside(&line, '"', '"');
        if (!isdigit_string(sp)) {
            log_message(WARNING, NOCLASS, MSG_BAD_INTVAL3s, __func__, line, token);
            return;
        }
    }

    if (0 <= index) {
        FREEUP(req->keyvalues[index].value);
        req->keyvalues[index].value = sp;
    } else {
        req->keyvalues[req->number].value = sp;
        req->number ++;
    }

    return;
} // end function handle_number_argument

static void
handle_option_argument(char *line, unsigned int option, unsigned int *options,
                       unsigned int *delOptions, const char *token)
{
    line = strip_spaces(line, FALSE);

    if (0 == strlen(line)) {
        log_message(WARNING, NOCLASS, MSG_WARN_NULLVAL2s, __func__, token);
        return;
    }

    if (is_stringtoken(line, "Y")) {
        *options |= option;
        *delOptions &= ~option;
        return;
    } else if (is_stringtoken(line, "SUB_RESET")) {
        *delOptions |= option;
        return;
    }

    log_message(WARNING, NOCLASS, MSG_BAD_BOOLVAL3s, __func__, line, token);

    return;
} // end function handle_option_argument

static bool_t
is_stringtoken(char *s, const char *tok)
{
    strip_spaces(s, FALSE);

    if (strncmp(s, tok, strlen(tok)) == 0) {
        char *p = s+strlen(tok);

        if (strlen(p) > 0) {
            return FALSE;
	}

        return TRUE;
    }

    return FALSE;
}

static int 
handle_other_files(char *token, char *line, struct request *req, unsigned int *deleteOptions)
{
    unsigned int i;
    if (0 == strcmp(token, "QP_SUBMIT_OTHER_FILES")) {

        if (is_stringtoken(line, "SUB_RESET")) {
            
            for (i=0; i<req->number; i++) {
                switch (req->keyvalues[i].key) {
                case REQUEST_KEY_SUBMIT_NUM_XFILE:
                    snprintf(req->keyvalues[i].value, 12, "%d", 0);
                    break;
                case REQUEST_KEY_SUBMIT_XFILE:
                    if (i == req->number -1) {
                        FREEUP(req->keyvalues[i].value);
                        req->number --;
                    } else {
                        req->keyvalues[i].key = req->keyvalues[req->number-1].key;
                        FREEUP(req->keyvalues[i].value);
                        req->keyvalues[i].value = req->keyvalues[req->number-1].value;
                        req->number --;
                    }

                    break;
                }
            }

            *deleteOptions |= SUBOPT_OTHER_FILES;
            return 0;
        }

        if (!isdigit_string(line)) {
            log_message(WARNING, NOCLASS, MSG_BAD_INTVAL3s, __func__, line, token);
            return -1;
        }

        for (i=0; i<req->number; i++) {
            switch (req->keyvalues[i].key) {
            case REQUEST_KEY_SUBMIT_NUM_XFILE:
                snprintf(req->keyvalues[i].value, 12, "%s", line);
                break;
            case REQUEST_KEY_SUBMIT_XFILE:
                if (i == req->number -1) {
                    FREEUP(req->keyvalues[i].value);
                    req->number --;
                } else {
                    req->keyvalues[i].key = req->keyvalues[req->number-1].key;
                    FREEUP(req->keyvalues[i].value);
                    req->keyvalues[i].value = req->keyvalues[req->number-1].value;
                    req->number --;
                }

                break;
            }
        }

        *deleteOptions &= SUBOPT_OTHER_FILES;

        return 0;
    } else {

        char *sp = token+strlen("QP_SUBMIT_OTHER_FILES_");
        char xfSeq[32];

        xfSeq[0] = '\0';
        strncat(xfSeq, sp, sizeof(xfSeq)-2);

        if(!isdigit_string(xfSeq)) {
            log_message(WARNING, NOCLASS, MSG_BAD_ENVAR2s, __func__, token);

            return -1;
        }

        int maxxf = 0;
        for (i=0; i<req->number; i++) {
            switch (req->keyvalues[i].key) {
            case REQUEST_KEY_SUBMIT_NUM_XFILE:
                maxxf = atoi(req->keyvalues[i].value);
            }
        }

        int v = atoi(xfSeq);
        if (v >= maxxf) {
            log_message(WARNING, NOCLASS, MSG_BAD_ENVAL3s, __func__, line, token);
            return -1;
        }

        req->keyvalues[req->number].key = REQUEST_KEY_SUBMIT_XFILE;
        req->keyvalues[req->number].value = copy_string(line);
        req->number ++;
    }

    return 0;
} // end function handle_other_files
